package cn.me.alphamq.nameserver.event.spi.listener.service;

import cn.me.alphamq.common.entity.resp.ErrorAccountResp;
import cn.me.alphamq.common.entity.resp.ServiceHeartBeatResp;
import cn.me.alphamq.common.sync.StopWatch;
import cn.me.alphamq.common.util.AttrUtil;
import cn.me.alphamq.common.util.NettyUtil;
import cn.me.alphamq.nameserver.cache.Container;
import cn.me.alphamq.common.msg.TcpMsg;
import cn.me.alphamq.common.enumeration.EventCodeEnum;
import cn.me.alphamq.nameserver.event.model.service.HeartBeatEvent;
import cn.me.alphamq.common.event.Listener;
import cn.me.alphamq.nameserver.store.ReplicationMsgQueueManager;
import cn.me.alphamq.nameserver.store.ServiceInstance;
import cn.me.alphamq.common.store.StopWatchManager;
import cn.me.alphamq.nameserver.store.ServiceInstanceManager;
import com.alibaba.fastjson2.JSON;
import io.netty.channel.ChannelHandlerContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 服务的心跳事件的监听器
 *
 * @author f
 */
public class HeartBeatListener implements Listener<HeartBeatEvent> {

    private static final Logger log = LoggerFactory.getLogger(HeartBeatListener.class);

    private final ServiceInstanceManager serviceInstanceManager = Container.getServiceInstanceManager();
    private final ReplicationMsgQueueManager replicationMsgQueueManager = Container.getReplicationMsgQueueManager();

    /**
     * 停表
     */
    private final StopWatch stopWatch = StopWatchManager.newStopWatch();

    // 本方法负责将存在的实例保存下来
    @Override
    public void onReceive(HeartBeatEvent event) throws Exception {
        ChannelHandlerContext ctx = event.getCtx();
        String msgId = event.getMsgId();

        // 如果服务实例的标识为空，则报错
        String attr = NettyUtil.getCtxId(ctx);
        if (attr == null) {
            ctx.writeAndFlush(new TcpMsg(EventCodeEnum.ERROR_ACCOUNT_RESP,
                    JSON.toJSONBytes(new ErrorAccountResp(msgId))));
            ctx.close();
            throw new IllegalAccessException("incorrect username or password");
        }

        // 将 attr 中的信息解析出来
        String[] split = attr.split(":");
        String brokerIp = split[0];
        int brokerPort = Integer.parseInt(split[1]);

        // 更新服务实例心跳的时间戳，并且更新 broker 的版本号
        ServiceInstance serviceInstance = serviceInstanceManager.get(brokerIp, brokerPort);
        serviceInstance.setLastHeartBeatTimestamp(System.currentTimeMillis());
        long version = event.getVersion();
        if (version > 0) {
            AttrUtil.putVersionToAttrs(serviceInstance.getAttrs(), version);
            Container.setCommitLogTopicModelList(event.getCommitLogTopicModelList());
            Container.setConsumeQueueOffsetModel(event.getConsumeQueueOffsetModel());
        }

        // 将修改的服务实例发给复制消息队列的管理者
        replicationMsgQueueManager.putShiftyServiceInstance(serviceInstance, stopWatch);

        // 给 broker 响应一条消息，表示心跳成功
        ctx.writeAndFlush(new TcpMsg(EventCodeEnum.HEART_BEAT_SUCCESS_RESP,
                JSON.toJSONBytes(new ServiceHeartBeatResp(msgId))));
//        log.info("心跳成功"); // TODO DELETE
    }
}
